package com.example.flasksale.listen;

import com.alibaba.fastjson.JSON;
import com.example.flasksale.entity.Order;
import com.example.flasksale.service.Consts;
import com.example.flasksale.service.OrderService;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = Consts.TOPIC_ORDER,
        consumerGroup = "order_consumer01",
        consumeMode = ConsumeMode.CONCURRENTLY,
        messageModel = MessageModel.CLUSTERING,
        secretKey = "*")
public class OrderListener implements RocketMQListener<Order> {

    @Autowired
    private OrderService orderService;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public void onMessage(Order order) {

        Long remain = redisTemplate.opsForValue().decrement(order.getProductId() + "");
        if (remain < 0) {
            // 多扣了
            Long increment = redisTemplate.opsForValue().increment(order.getProductId() + "");
            System.out.println("没库存了：" + increment);
            return;
        }

        // 创建订单
        orderService.add(order);

        System.out.println("order id: " + order.getId());

        // 发送消息到超时对列
        Message message = new Message(Consts.TOPIC_ORDER_TIMEOUT, JSON.toJSONString(order).getBytes());
        org.springframework.messaging.Message springMessage = RocketMQUtil.convertToSpringMessage(message);
        rocketMQTemplate.asyncSend(Consts.TOPIC_ORDER_TIMEOUT, springMessage, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {

            }

            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
            }
        }, 1_000, 4);

    }
}
